package com.fanxuankai.boot.monitor.confluent;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.fanxuankai.boot.monitor.Message;
import com.fanxuankai.boot.monitor.Monitor;
import com.fanxuankai.boot.monitor.Notifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * @author fanxuankai
 */
public class ConfluentConnectorMonitor implements Monitor {
    @Resource
    private ConfluentConnectorProperties properties;
    @Resource
    private RestTemplate restTemplate;
    @Resource
    private Notifier<Message> notifier;
    private static final Logger LOGGER = LoggerFactory.getLogger(ConfluentConnectorMonitor.class);

    @Override
    public String title() {
        return properties.getTitle();
    }

    @Override
    public String cron() {
        return properties.getCron();
    }

    @Override
    public void exec() {
        for (ConfluentConnectorProperties.ConnectorConfig config : properties.getConfigs()) {
            String url = config.getClusters() + config.getUrl();
            String json = restTemplate.getForObject(url, String.class);
            JSONObject jsonObject = JSONObject.parseObject(json);
            List<String> contents = new ArrayList<>(jsonObject.size());
            for (String key : jsonObject.keySet()) {
                JSONObject value = jsonObject.getJSONObject(key);
                JSONObject status = value.getJSONObject("status");
                String connectorState = status.getJSONObject("connector").getString("state");
                if (!StrUtil.equals(connectorState, config.getRunningState())) {
                    contents.add(String.format("%s : %s", key, connectorState));
                    tryRestart(config, key);
                    continue;
                }
                JSONArray tasks = status.getJSONArray("tasks");
                for (int i = 0; i < tasks.size(); i++) {
                    JSONObject task = tasks.getJSONObject(i);
                    String taskState = task.getString("state");
                    if (!StrUtil.equals(taskState, config.getRunningState())) {
                        contents.add(String.format("%s task %s: %s", key, i, taskState));
                        tryRestart(config, key);
                        break;
                    }
                }
            }
            if (!contents.isEmpty()) {
                Message message = new Message();
                message.setTitle(properties.getTitle());
                message.setSubtitle(Optional.ofNullable(properties.getSubtitle())
                        .orElse(config.getClusters()));
                message.setContents(contents);
                String description = "共计 " + contents.size() + " 个连接器异常";
                if (config.isAutoRestart()) {
                    description += "，已尝试重启!";
                }
                message.setDescription(description);
                notifier.notice(message);
            }
        }
    }

    private void tryRestart(ConfluentConnectorProperties.ConnectorConfig config, String connector) {
        if (config.isAutoRestart()) {
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);
            HttpEntity<String> entity = new HttpEntity<>(null, headers);
            String r = restTemplate.postForObject(config.getClusters() + config.getRestartUrl(),
                    entity, String.class, connector);
            LOGGER.info("尝试重启 {}: {}", connector, r);
        }
    }
}
